/* ***** BEGIN LICENSE BLOCK *****
 * Version: MPL 1.1/GPL 2.0/LGPL 2.1/commercial
 *
 * The contents of this file are subject to the Mozilla Public License Version
 * 1.1 (the "License"); you may use this file in compliance with the License.
 * Alternatively, if you can't copy with the License, namely if you do not
 * want to provide source code of your modifications to this file, you may
 * negotiate a commercial licence with the initial developer.
 * You may obtain a copy of the License at
 * http://www.mozilla.org/MPL/
 *
 * Software distributed under the License is distributed on an "AS IS" basis,
 * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
 * for the specific language governing rights and limitations under the
 * License.
 *
 * The Original Code is COID.
 *
 * The Initial Developer of the Original Code is
 * PosAm.
 * Portions created by the Initial Developer are Copyright (C) 2003
 * the Initial Developer. All Rights Reserved.
 *
 * Contributor(s):
 * Brano Kemen
 *
 * Alternatively, the contents of this file may be used under the terms of
 * either the GNU General Public License Version 2 or later (the "GPL"), or
 * the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
 * in which case the provisions of the GPL or the LGPL are applicable instead
 * of those above. If you wish to allow use of your version of this file only
 * under the terms of either the GPL or the LGPL, and not to allow others to
 * use your version of this file under the terms of the MPL, indicate your
 * decision by deleting the provisions above and replace them with the notice
 * and other provisions required by the GPL or the LGPL. If you do not delete
 * the provisions above, a recipient may use your version of this file under
 * the terms of any one of the MPL, the GPL or the LGPL.
 *
 * ***** END LICENSE BLOCK ***** */

#include "coid/comm/pthreadx.h"
#include "coid/comm/binstream/textstream.h"
#include "server.h"
#include "acceptor.h"
#include "coid/coid.h"

COID_NAMESPACE_BEGIN


////////////////////////////////////////////////////////////////////////////////
/*bool ConnectionCoid::destroy()
{
    //signal termination
    _flags |= fEXIT;
    sched_yield();

    if (!(_flags & fSTOPPED)) {
        SRVROOTA.add_to_dead_list(get_id());
        return false;
    }

    return true;
}*/

thread_key   ConnectionCoid::_thread_key;

////////////////////////////////////////////////////////////////////////////////
/// called before the children nodes are detached
opcd ConnectionCoid::predestroy()
{
    //signal termination
    _flags |= fEXIT;

#ifndef SYSTYPE_WIN
    sched_yield();
#else
    sysMilliSecondSleep(0);
#endif

    GUARDME_REG;
    free_threads();

    return 0;
}

////////////////////////////////////////////////////////////////////////////////
ConnectionCoid::~ConnectionCoid()
{
}

////////////////////////////////////////////////////////////////////////////////
uint ConnectionCoid::add_service_inst( ServiceInstanceCoid* svci )
{
    GUARDME_REG;
    *_inst.add(1) = svci;

    if( svci->_owner != _id )
        SRVROOTA._objmgr.object_hook( get_id(), svci->get_id() );
    else
        SRVROOTA._objmgr.object_hook( svci->get_id(), get_id() );
    return uint(_inst.size() - 1);
}

////////////////////////////////////////////////////////////////////////////////
opcd ConnectionCoid::list_cmds( const token& cmd, dynarray<charstr>& cmds ) const
{
    if( cmd.is_empty() )
    {
        cmds.push("connect SERVICENAME");
        cmds.push("connect -s SERVICENAME");
        cmds.push("connect -s ID[:NAME]");

        cmds.push("lsl");

        return CoidNode::list_cmds( cmd, cmds );
    }
    else if( cmd == "connect" )
    {
        cmds.push("connect SERVICENAME\n\tconnect to new instance of service SERVICENAME");
        cmds.push("connect -s SERVICENAME\n\tshared connect to autospawned instance of service SERVICENAME");
        cmds.push("connect -s ID[:NAME]\n\tshared connect to object with id ID, optionally check for name NAME");

        cmds.push("lsl\n\tlist slots (bound services) of this connection object");
    }
    else
        return CoidNode::list_cmds( cmd, cmds );

    return 0;
}

////////////////////////////////////////////////////////////////////////////////
opcd ConnectionCoid::dispatch_cmd( const token& cmd, command_tokens& arg, txtstream& reply )
{
    if( cmd == "connect" )
    {
        opcd e = 0;
        if (arg[0].is_empty())
        {
            reply << "missing arguments" << EOM;
            return 0;
        }
        else if( arg[0] == "-s"  ||  arg[0] == "--shared" )
        {
            token t = arg[1];
            if( t.is_empty() )
            {
                reply << "missing object id or class name" << EOM;
                return 0;
            }

            uint id = t.touint_and_shift();

            token n = token::empty();
            if( !t.is_empty() )
                n = arg[1];


            NODEREF<ServiceInstanceCoid> psi;
            e = SRVROOTA._objmgr.serviceinst_find( id, n, version(UMAX32), psi );

            if(!e)
            {
                binstreambuf bib( arg[2]._ptr );
                textstream txp( bib );
                
                e = psi->dpx_authentify_shared(txp,0);
                if( e == ersNOT_IMPLEMENTED )
                    e = 0;
            }

            if(e)
            {
                reply << "error connecting service: ";
                if( n.is_empty() )
                    reply << "id: " << id;
                else
                    reply << "class: " << n;
                reply << " error: " << opcd_formatter(e) << EOM;
            }
            else
                add_service_inst( psi.ptr() );
        }
        else
        {
            NODEREF<ServiceCoid> psi;
            e = SRVROOTA._objmgr.service_find( arg[0], version(UMAX32), psi );

            if(!e)
            {
                binstreambuf bib( arg[1]._ptr );
                textstream txp( bib );
                e = psi->spawn_instance( txp );
            }

            if(e)
                reply << "error connecting service: " << arg[0] << " error: " << opcd_formatter(e) << EOM;
        }

        return 0;
    }
    else if( cmd == "lsl" )
    {
        for( uint i=0; i<_inst.size(); ++i )
        {
            if( _inst[i] )
            {
                charstr t;
                reply << ">" << i << "\t (" << _inst[i]->_id << ") " << _inst[i]->class_name() << "-" << _inst[i]->_desc->_version.get_version(t);
                reply << " " << _inst[i]->_name;

                if( _inst[i]->_owner != _id )
                    reply << "\t(shared)";

                reply << "\n";
            }
        }
        return 0;
    }

    return CoidNode::dispatch_cmd( cmd, arg, reply );    
}

////////////////////////////////////////////////////////////////////////////////
opcd ConnectionCoid::attach_service_inst( uint soid, const token& name, const version& svcver, binstream& arg )
{
    NODEREF<ServiceInstanceCoid> psvci;

    opcd e = SRVROOTA._objmgr.serviceinst_find( soid, name, svcver, psvci );
    if(!e)
    {
        e = psvci->dpx_authentify_shared(arg,0);
        if(!e)
            add_service_inst( psvci );
    }

    return e;
}

////////////////////////////////////////////////////////////////////////////////
opcd ConnectionCoid::loop()
{
    ThreadMgr::set_thread_name( thread::self(), "dispatch loop" );
    ThreadMgr::set_thread_tempid( thread::self(), _id );

    try
    {
        set_active_conn();
        dispatch_loop();
        clear_active_conn();
    }
    catch( thread::Exception& )
    {
        clear_active_conn();
        throw;
    }
    catch( opcd e )
    {
        clear_active_conn();
        return e;
    }
    catch( ... )
    {
        clear_active_conn();
        return ersEXCEPTION;
    }

    return 0;
}

////////////////////////////////////////////////////////////////////////////////
opcd ConnectionCoid::dispatch_loop()
{
    uint obj;
    uint method;

    try
    {
        while( !should_exit()  &&  !thread::self().should_cancel() )
        {
            obj = UMAX32;
            method = UMAX32;

            _bin->set_timeout(PACKET_TIMESLICE);
            opcd e = _bin->read_type( &method );

            if( e == ersDISCONNECTED )
            {
                _flags |= fEXIT;
                break;
            }
            if( e == ersTIMEOUT )
            {
                if(_flags & fEXITWAIT)
                {
                    if( _timetoexit == 0 )
                    {
                        _flags |= fEXIT;
                        break;
                    }

                    --_timetoexit;
                }

                continue;
            }
            else if(e)
            {
                //hm
                _bin->acknowledge(true);
                continue;
                //_flags |= fEXIT;
                //break;
            }

            //if _timeout == 0, fEXITWAIT was set for one event, otherwise reset the timeout counter and keep the flag
            if(!_timeout)
                _flags &= fEXITWAIT;
            else
                _timetoexit = _timeout / PACKET_TIMESLICE;

            _bin->set_timeout(0);

            
            if( method == UMAX32 )  break;
            if( method == StdProtocolMethod::CONNECT )
            {
                NODEREF<CoidAcceptorCoid> coidaccp;
                NODEREF<ConnectionCoid> conn;

                opcd e = SRVROOTA._objmgr.get_node( get_id(), conn );
                if(!e)
                e = SRVROOTA._objmgr.get_node( SRVROOTA._accpcoid_nodeid, coidaccp );
                if(e)
				{
					_bin->acknowledge(true);
					(*_bin) << ersFE_CHANNEL << BINSTREAM_FLUSH;
					continue;
				}

                coidaccp->attach_to_connection( conn, *_bin, _client_addr );
				continue;
            }

            (*_bin) >> obj;


            e = dispatch_call( *_bin, obj, method, 0 );
            if(e)
                return e;
        }
    }
    catch( thread::Exception& )
    {
        post_trace(1) << "dispatch was aborted by system exception (thread::Exception)" << EOM;
        //SRVROOTA.detach (this, 2);
        throw;
    }
//#ifdef COID_CATCH_GLOBAL_EXCEPTIONS
    catch( opcd e )
    {
        binstream& b = post_critical(0);
        b << "unhandled exception occured in the dispatch thread of this connection (from:";
        charstr clia;
        if( _nick.is_empty() )
            b << get_client_addr(clia);
        else
            b << _nick << " - " << get_client_addr(clia);
        b << "). Error: " << opcd_formatter(e) << "\n";

        if( obj != UMAX32 )
            dump_method_desc( b, obj, method );
        b << EOM;
    }
    catch (...)
    {
        binstream& b = post_critical(0);
        b << "unhandled exception occured in the dispatch thread of this connection (from:";
        charstr clia;
        if( _nick.is_empty() )
            b << get_client_addr(clia);
        else
            b << _nick << " - " << get_client_addr(clia);
        b << "): crash bum bang i'm dead (hard exception, probably an access violation)\n";
        
        if( obj != UMAX32 )
            dump_method_desc( b, obj, method );
        b << EOM;
    }

    return 0;
}

////////////////////////////////////////////////////////////////////////////////
opcd ConnectionCoid::dispatch_call( binstream& bin, uint obj, uint method, metastream* meta )
{
    GUARDME_REG;
    
    if( obj >= _inst.size() )
    {
        bin >> BINSTREAM_ACK_EAT;
        bin << ersFE_NOT_ATTACHED_SVC;
        bin.flush();
        post_error(0) << "querying a nonexistent service index" << EOM;
    }
    else if( _inst[obj] == 0 )
    {
        bin >> BINSTREAM_ACK_EAT;
        bin << ersFE_SVC_ALREADY_DELETED;
        bin.flush();
        post_error(0) << "querying an already deleted service index " << obj << EOM;
    }
    else if( method & 0x80000000 )
    {
        LNID k;

        switch (method) {
        case StdProtocolMethod::PING:
            {
                bin.acknowledge();
                bin << opcd(0);
                bin.flush();
                break;
            }
        case StdProtocolMethod::DISCONNECT_AND_WAIT:
            {
                uint timeout;
                bin >> timeout;
                _timetoexit = timeout / PACKET_TIMESLICE;
                bin.acknowledge();
                bin << opcd(0);
                bin.flush();

                post_trace(1) << "disconnecting bcon object with timeout " << timeout << "ms" << EOM;

                _bin.destroy();
                _flags |= fEXITWAIT;
                return 0;//1;
            }

        case StdProtocolMethod::DISCONNECT:
            {
                bin.acknowledge();
                bin << opcd(0);
                bin.flush();

				if( _inst[obj]->_owner == _id )
					_inst[obj]->request_detach();
                _inst[obj] = 0;

                //exit if no instances served any longer
                {
                    uint i;
                    for( i=0; i<_inst.size(); ++i )
                        if( _inst[i] != 0 )  break;

                    if( i>=_inst.size() )
                        _flags |= fEXIT;
                }
                
                return (_flags & fEXIT) ? ersEXIT : opcd(0);
            }

        case StdProtocolMethod::GET_INSTANCE_ADDRESS:
            {
                bin.acknowledge();
                netaddr oa;  uint objid;
                opcd e = _inst[obj]->get_object_address(oa,&objid);
                bin << e;
                if (!e)
                    bin << oa << objid;
                bin.flush();
                break;
            }

        case StdDispatchMethod::SPAWN:
            {
                bin.acknowledge();

                int respawn = 0;
                if( (_inst[obj]->_flags & ServiceInstanceCoid::fALLOW_SHARED)
                    && !(_inst[obj]->_flags & ServiceInstanceCoid::fAUTOINSTANCE) )
                {
                    post_info(1) << "cannot spawn, shared connections already permitted"
                        << _inst[obj]->class_name() << ")" << EOM;
                    break;
                }

                try
                {
                    _inst[obj]->dispatch( method, bin, meta );
                }
                catch( thread::Exception& )   { throw; }
                catch( opcd e )
                {
                    post_critical(0) << "exception in main thread: "
                        << opcd_formatter(e) << ", respawning" << EOM;
                    ++respawn;
                }
                catch(...)
                {
                    post_critical(0) << "exception in main thread, respawning" << EOM;
                    ++respawn;
                }

                //if(respawn)
                //    SRVROOTA.ask_run_update_plglist();

                if( _inst[obj]->_flags & ServiceInstanceCoid::fAUTOINSTANCE )
                {
                    post_info(1) << "autospawned service returned from its thread, terminating the autospawn instance ("
                        << _inst[obj]->class_name() << ")" << EOM;
                    _flags |= fEXIT;
                    return ersEXIT;
                }
                break;
            }

        case StdProtocolMethod::SET_CONN_TIMEOUT:
            {
                bin >> _timeout;
                bin.acknowledge();
                bin << opcd(0);
                bin.flush();
            }
            break;

        default:
            {
                bin >> BINSTREAM_ACK_EAT;
                bin << ersFE_UNKNOWN_METHOD;
                bin.flush();
                post_error(0) << "unrecognized system event request: " << method << EOM;
            }
        }
    }
    else if( _inst[obj]->_flags & ServiceInstanceCoid::fAUTHENTIFIED )
    {
        if( 0 == _inst[obj]->_desc->is_implemented(method) )
        {
            bin >> BINSTREAM_ACK_EAT;
            bin << ersFE_UNKNOWN_METHOD;
            bin.flush();
            post_error(0) << "unknown method call: " << method << EOM;
        }
        else if( _inst[obj]->_desc->is_unlocked(method) )
        {
            _inst[obj]->dispatch( method, bin, meta );
        }
        else
        {
            GUARD g( *_inst[obj], _inst[obj]->_desc->is_const(method) != 0 );

            _inst[obj]->dispatch( method, bin, meta );
        }
    }
    else {
        bin >> BINSTREAM_ACK_EAT;
        bin << ersFE_NOT_AUTHENTIFIED;
        bin.flush();
        post_error(0) << "not authentified, querying method: " << method << EOM;
    }

    return 0;
}

////////////////////////////////////////////////////////////////////////////////
void ConnectionCoid::set_active_conn()
{
    _thread_key.set(this);
}

////////////////////////////////////////////////////////////////////////////////
void ConnectionCoid::clear_active_conn()
{
    _thread_key.set(0);
}

////////////////////////////////////////////////////////////////////////////////
ConnectionCoid* ConnectionCoid::get_active_conn()
{
    return (ConnectionCoid*)_thread_key.get();
}



COID_NAMESPACE_END
